package com.jhc.drone.communication.ws;


import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.json.JSONObject;
import com.google.gson.Gson;
import com.jhc.drone.communication.datagram.command.Command;
import com.jhc.drone.communication.enums.FlyModeEnum;
import com.jhc.drone.communication.enums.WsReqEnum;
import com.jhc.drone.communication.network.cache.DroneCache;
import com.jhc.drone.communication.network.cache.WebsocketCache;
import com.jhc.drone.communication.processor.MessageProcessor;
import com.jhc.drone.communication.processor.PackMessageProcessor;
import com.jhc.drone.communication.properties.ProtocolProperties;
import com.jhc.drone.communication.vo.CommandVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author lcp
 * @date 2024年12月2024/12/11 10:21
 * @describe ws
 */
@ServerEndpoint("/websocket/{flag}/{sn}")
@Component
@Slf4j
public class WsEndpoint implements ApplicationContextAware {

    private static final Long OUT_TIME_INTERVAL = 30L;

    private static final Gson gson = new Gson();

    public static MessageProcessor messageProcessor;

    private static final ExecutorService executorService = Executors.newCachedThreadPool();


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        ProtocolProperties protocolEnum = applicationContext.getBean(ProtocolProperties.class);
        messageProcessor = new PackMessageProcessor(protocolEnum.getProtocol());
    }

    /**
     * @param session 会话
     * @describe 描述：websocket建立连接
     * @author lcp
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("sn") String sn, @PathParam("flag") Integer flag) {
        if (flag == 0) {
            WebsocketCache.SESSIONS.add(session);
        } else {
            if (WebsocketCache.SESSION_MAP.containsKey(sn)) {
                WebsocketCache.SESSION_MAP.get(sn).add(session);
            } else {
                ConcurrentHashSet<Session> objects = new ConcurrentHashSet<>();
                objects.add(session);
                WebsocketCache.SESSION_MAP.put(sn, objects);
            }
        }
        log.info("onOpen【{}】", session.getId());
    }

    /**
     * 收到客户端消息的回调方法
     *
     * @param message 客户端传过来的消息
     * @param session 对应的session
     */
    @OnMessage
    public void onMessage(String message, Session session) throws Exception {
        CompletableFuture.runAsync(() -> {
            try {
                log.info("receive message【{}】:{}", session.getId(), message);
                if (!message.isEmpty()) {
                    CommandVo commandVo = gson.fromJson(message, CommandVo.class);
                    if (commandVo.getWsReqEnum() == WsReqEnum.MOVE_CONTROLLER) {
                        // 通知上一次线程退出循环
                        WebsocketCache.IS_LOOP1RING.set(false);
                        // 阻塞等待上一次线程结束任务
                        String sn = WebsocketCache.getSn(session);
                        commandVo.setSn(sn);
                        commandVo.setSysId(DroneCache.getSysId(sn));
                        startRcLoop(commandVo);
                    } else if (commandVo.getWsReqEnum() == WsReqEnum.RELEASE_RC) {
                        WebsocketCache.IS_LOOP1RING.set(false);
                    } else {
                        String sn = WebsocketCache.getSn(session);
                        commandVo.setSn(sn);
                        loop(commandVo);
                    }
                }
            } catch (Exception e) {
                throw new RuntimeException("子线程被中断", e);
            }
        }, executorService).exceptionally(ex -> {
            // 在这里处理异常，记录日志或者发送 WebSocket 错误消息
            log.error("子线程异常: {}", ex.getMessage(), ex);
            JSONObject entries = new JSONObject();
            entries.set("errorCode", "500");
            entries.set("errorMessage", "命令执行发生了异常:" + ex.getMessage());
            try {
                session.getBasicRemote().sendText(entries.toString());
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            return null;
        });
    }

    private synchronized void startRcLoop(CommandVo commandVo) {
        // 成功拿到锁资源，重新赋值保证循环不会结束
        WebsocketCache.IS_LOOP1RING.set(true);
        while (WebsocketCache.IS_LOOP1RING.get()) {
            if(DroneCache.FLY_MODE_MAP.containsKey(commandVo.sn) && DroneCache.FLY_MODE_MAP.get(commandVo.sn) != FlyModeEnum.POINT_HOLD_MODE){
                break;
            }
            if (!WebsocketCache.IS_LOOP1RING.get()) {
                break;
            }
            loop(commandVo);
            try {
                Thread.sleep(OUT_TIME_INTERVAL);
            } catch (Exception e) {
                log.error(e.getMessage());
            }
        }
    }

    private void loop(CommandVo commandVo) {
        byte[] bytes = Command.buildCommand(commandVo);
        messageProcessor.send(bytes, commandVo.sn);
    }

    /**
     * 发生错误的回调方法
     *
     * @param session 会话
     * @param error   异常
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("发生错误【{}】", session.getId(), error);
    }

    /**
     * 关闭连接的回调方法
     */
    @OnClose
    public void onClose(Session session) {
        WebsocketCache.SESSIONS.remove(session);
        WebsocketCache.SESSION_MAP.forEach((key, value) -> value.remove(session));
    }

    /**
     * 群发送消息
     *
     * @param message 消息
     */
    public static void sendMessage(WsDto message) {
        WebsocketCache.SESSION_MAP.forEach((k, v) -> {
            if (Objects.equals(k, message.getSn())) {
                v.stream().parallel().forEach(session -> {
                    jsonWrapper(session, message);
                });
            }
        });
    }

    public static void sendMessage(JSONObject message) {
        WebsocketCache.SESSION_MAP.forEach((k, v) -> {
            if (Objects.equals(k, message.get("sn"))) {
                v.stream().parallel().forEach(session -> {
                    jsonWrapper(session, message);
                });
            }
        });
    }

    public static void broadMessage(JSONObject message) {
        WebsocketCache.SESSIONS.stream().parallel().forEach(session -> {
            jsonWrapper(session, message);
        });
    }

    public static void sendAll(JSONObject message) {
        sendMessage(message);
        broadMessage(message);
    }

    private static void jsonWrapper(Session session, WsDto message) {
        try {
            JSONObject entries = new JSONObject();
            entries.set("type", "1");
            entries.set("data", message);
            session.getBasicRemote().sendText(gson.toJson(entries));
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private static void jsonWrapper(Session session, JSONObject message) {
        try {
            JSONObject entries = new JSONObject();
            entries.set("type", "1");
            entries.set("data", message);
            session.getBasicRemote().sendText(gson.toJson(entries));
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }
}
